package rabbitmq

import (
	"errors"

	"git.100tal.com/wangxiao_dahai_backend/arch-nova/middleware"
	"github.com/assembla/cony"
	"github.com/streadway/amqp"
)

// 单生产者
func (mq *RabbitMq) NewProducer(e *Exchange, q *Queue, routeKey string) error {
	if e.Name == "" || e.Type == "" {
		return errors.New("exchange name or kind error")
	}
	if !checkExchangeType(e.Type) {
		return errors.New("exchange type error")
	}

	exc := cony.Exchange{}
	exc.Name = e.Name
	exc.Kind = e.Type
	exc.Durable = e.Durable
	exc.AutoDelete = e.AutoDelete
	exc.Args = e.Args

	if q == nil {
		mq.cli.Declare([]cony.Declaration{
			cony.DeclareExchange(exc),
		})
	} else {
		que := &cony.Queue{}
		que.Name = q.Name
		que.Durable = q.Durable
		que.AutoDelete = q.AutoDelete
		que.Exclusive = q.Exclusive
		que.Args = q.Args

		bnd := cony.Binding{
			Queue:    que,
			Exchange: exc,
			Key:      routeKey,
		}

		mq.cli.Declare([]cony.Declaration{
			cony.DeclareQueue(que),
			cony.DeclareExchange(exc),
			cony.DeclareBinding(bnd),
		})
	}

	pbl := cony.NewPublisher(exc.Name, routeKey)
	mq.cli.Publish(pbl)

	// Launch a go routine and publish a message.
	// "Publish" is a blocking method this is why it
	// needs to be called in its own go routine.
	//
	go func() {
		defer func() {
			if err := recover(); err != nil {
				middleware.LogError(nil, "[RecoveryPanic] [RabbitMq Producer] "+err.(string))
			}
		}()

		for {
			select {
			case body := <-mq.Message:
				err := pbl.Publish(amqp.Publishing{
					Body: []byte(body),
				})
				if err != nil {
					e := errors.New("mq_producer publish error: " + err.Error())
					mq.Error <- e
				}
			}
		}
	}()

	// Client loop sends out declarations(exchanges, queues, bindings
	// etc) to the AMQP server. It also handles reconnecting.
	//
	go func() {
		defer func() {
			if err := recover(); err != nil {
				middleware.LogError(nil, "[RecoveryPanic] [RabbitMq Producer] "+err.(string))
			}
		}()

		for mq.cli.Loop() {
			select {
			case err := <-mq.cli.Errors():
				e := errors.New("mq_producer_client error: " + err.Error())
				mq.Error <- e

			case blocked := <-mq.cli.Blocking():
				e := errors.New("mq_producer_client is blocked: " + blocked.Reason)
				mq.Error <- e
			}
		}
	}()

	return nil
}

// 发送消息
func (mq *RabbitMq) Send(message string) {
	mq.Message <- message
}

// 单生产者简化版接口 direct
// 无须配置exchange queue等
// 默认持久化
func (mq *RabbitMq) NewProducerDirect(exchangeName, queueName, routeKey string) error {
	e := NewExchange(
		ExchangeName(exchangeName),
		ExchangeType(EXCHANGE_TYPE_DIRECT),
		ExchangeDurable(true),
	)
	q := NewQueue(
		QueueName(queueName),
		QueueDurable(true),
	)
	err := mq.NewProducer(e, q, routeKey)
	if err != nil {
		return err
	}

	return nil
}

// 单生产者简化版接口 fanout
// 无须配置exchange queue等
// 默认持久化
func (mq *RabbitMq) NewProducerFanout(exchangeName, queueName, routeKey string) error {
	e := NewExchange(
		ExchangeName(exchangeName),
		ExchangeType(EXCHANGE_TYPE_FANOUT),
		ExchangeDurable(true),
	)
	q := NewQueue(
		QueueName(queueName),
		QueueDurable(true),
	)
	err := mq.NewProducer(e, q, routeKey)
	if err != nil {
		return err
	}

	return nil
}

// 多生产者包装
type ProducerWrapper struct {
	pbl     *cony.Publisher
	e       *cony.Exchange
	q       *cony.Queue
	k       string
	Message chan string // 发送的消息
}

func (mq *MqWrapper) NewProducerWrapper(e *Exchange, q *Queue, routeKey string) (*ProducerWrapper, error) {
	if e.Name == "" || e.Type == "" {
		return nil, errors.New("exchange name or type empty")
	}
	if !checkExchangeType(e.Type) {
		return nil, errors.New("exchange type error")
	}

	exc := cony.Exchange{}
	exc.Name = e.Name
	exc.Kind = e.Type
	exc.Durable = e.Durable
	exc.AutoDelete = e.AutoDelete
	exc.Args = e.Args

	que := &cony.Queue{}
	if q == nil {
		mq.cli.Declare([]cony.Declaration{
			cony.DeclareExchange(exc),
		})
	} else {
		que.Name = q.Name
		que.Durable = q.Durable
		que.AutoDelete = q.AutoDelete
		que.Exclusive = q.Exclusive
		que.Args = q.Args

		bnd := cony.Binding{
			Queue:    que,
			Exchange: exc,
			Key:      routeKey,
		}

		mq.cli.Declare([]cony.Declaration{
			cony.DeclareQueue(que),
			cony.DeclareExchange(exc),
			cony.DeclareBinding(bnd),
		})
	}

	pbl := cony.NewPublisher(exc.Name, routeKey)
	mq.cli.Publish(pbl)

	p := &ProducerWrapper{
		pbl:     pbl,
		e:       &exc,
		q:       que,
		Message: make(chan string),
	}

	// Launch a go routine and publish a message.
	// "Publish" is a blocking method this is why it
	// needs to be called in its own go routine.
	//
	go func() {
		defer func() {
			if err := recover(); err != nil {
				middleware.LogError(nil, "[RecoveryPanic] [RabbitMq Producer] "+err.(string))
			}
		}()

		for {
			select {
			case body := <-p.Message:
				err := p.pbl.Publish(amqp.Publishing{
					Body: []byte(body),
				})
				if err != nil {
					e := errors.New("mq_producer_wrapper publish error: " + err.Error())
					mq.Error <- e
				}
			}
		}
	}()

	return p, nil
}

// 发送消息
func (p *ProducerWrapper) Send(message string) {
	p.Message <- message
}

func (p *ProducerWrapper) GetExchangeName() string {
	if p.e != nil {
		return p.e.Name
	}

	return ""
}

func (p *ProducerWrapper) GetQueueName() string {
	if p.q != nil {
		return p.q.Name
	}

	return ""
}

func (p *ProducerWrapper) GetRoutingKey() string {
	return p.k
}
